package cn.gwm.flink.streaming.ods;

import cn.gwm.flink.streaming.factory.DataHubSourceFactory;
import cn.gwm.flink.streaming.function.process.OriginalDataProcessFunction;
import cn.gwm.flink.streaming.ods.model.ModelCanEnum;
import cn.gwm.flink.streaming.ods.model.ModelGpsEnum;
import cn.gwm.utils.ConfigLoader;
import cn.gwm.utils.DateTimeUtil;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

import java.io.IOException;
import java.util.Map;

/**
 * @ClassName OdsPreTask
 * @Author xzh
 * @Date 2022/11/25 16:15
 **/
public class OdsPreTask {

    public static SingleOutputStreamOperator<String> canStream(StreamExecutionEnvironment env, Map<String, OutputTag<String>> tagMap) throws IOException {
        return env.addSource(DataHubSourceFactory.getDatahubSourceFunction(ModelCanEnum.class))
                .process(new OriginalDataProcessFunction(tagMap)).returns(Types.STRING);
    }
    public static SingleOutputStreamOperator<String> gpsStream(StreamExecutionEnvironment env, Map<String, OutputTag<String>> tagMap) throws IOException {
        return env.addSource(DataHubSourceFactory.getDatahubSourceFunction(ModelGpsEnum.class))
                .process(new OriginalDataProcessFunction(tagMap)).returns(Types.STRING);
    }

    public static SingleOutputStreamOperator<String> defindStream(StreamExecutionEnvironment env, Map<String, OutputTag<String>> tagMap,String subId) throws IOException {
        return defindStream(env,tagMap,subId,ConfigLoader.get("all.receive.endTime"));
    }

    public static SingleOutputStreamOperator<String> defindStream(StreamExecutionEnvironment env, Map<String, OutputTag<String>> tagMap,String subId,String endTime) throws IOException {
        return env.addSource(DataHubSourceFactory.getDatahubSourceFunction(ConfigLoader.get("dataHub.topic.can"), subId, DateTimeUtil.defaultFormatPare(endTime)))
                .process(new OriginalDataProcessFunction(tagMap)).returns(Types.STRING);
    }

    public static SingleOutputStreamOperator<String> defindGpsStream(StreamExecutionEnvironment env, Map<String, OutputTag<String>> tagMap,String subId) throws IOException {
        return env.addSource(DataHubSourceFactory.getDatahubSourceFunction(ConfigLoader.get("dataHub.topic.gps"), subId, DateTimeUtil.defaultFormatPare(ConfigLoader.get("all.receive.endTime"))))
                .process(new OriginalDataProcessFunction(tagMap)).returns(Types.STRING);
    }

    /**
     *@Author xzh
     *@Description gps 步数据获取源数据
     *@Date 2023/2/27 14:01
     *@Param [env, tagMap, subId]
     *@return org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<java.lang.String>
     **/
    public static SingleOutputStreamOperator<String> getGspStream(StreamExecutionEnvironment env, Map<String, OutputTag<String>> tagMap) throws IOException {
        return env.addSource(DataHubSourceFactory.getDatahubSourceFunction("ntsp_t5_pdcu_fault_detection_monitor_data", "1667706214906UHXCP",
                DateTimeUtil.defaultFormatPare("2023-12-06 19:00:00")))
                .process(new OriginalDataProcessFunction(tagMap)).returns(Types.STRING);
    }
}
